package com.thiswhyme.stream.rabbit.producer.service;

import com.thiswhyme.stream.rabbit.producer.channel.OutputChannel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @description:
 * @author: thiswhyme
 * @date: 2020/9/16
 */
@Service
public class BizService {

    @Autowired
    private OutputChannel outputChannel;

    /**
     * 发送普通消息
     */
    public void sendLogMessage() {
        outputChannel.logOutputChannel().send(MessageBuilder.withPayload("log message").build());
    }

    /**
     * 发送分区消息
     */
    public void sendLPartitionMessage() {
        outputChannel.partitionOutputChannel().send(MessageBuilder.withPayload("partition message 1").setHeader("partitionKey", 1).build());
        outputChannel.partitionOutputChannel().send(MessageBuilder.withPayload("partition message 2").setHeader("partitionKey", 2).build());
    }

    /**
     * 发送延时消息,rabbitmq需要安装rabbitmq_delayed_message_exchange插件
     * 参见 https://www.rabbitmq.com/community-plugins.html
     */
    public void sendDelayMessage() {
        String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
        outputChannel.delayOutputChannel().send(MessageBuilder.withPayload("1.延时10s，delay message: " + date).setHeader("x-delay", 10000).build());
        outputChannel.delayOutputChannel().send(MessageBuilder.withPayload("2.延时5s，delay message: " + date).setHeader("x-delay", 5000).build());
        outputChannel.delayOutputChannel().send(MessageBuilder.withPayload("3.延时0s，delay message: " + date).setHeader("x-delay", 0).build());
        outputChannel.delayOutputChannel().send(MessageBuilder.withPayload("4.不延时，delay message: " + date).build());
    }

    /**
     * 发送普通消息（消费者手动ack）
     */
    public void sendManualMessage(String msg) {
        outputChannel.manualOutputChannel().send(MessageBuilder.withPayload(msg).build());
    }

    /**
     * 发送路由消息（topic）消息
     */
    public void sendRouteOutputMessage() {
        outputChannel.topicRouteOutputChannel().send(MessageBuilder.withPayload("bizType1 message").setHeader("bizType", "bizType1").build());
        outputChannel.topicRouteOutputChannel().send(MessageBuilder.withPayload("bizType2 message").setHeader("bizType", "bizType2").build());
        outputChannel.topicRouteOutputChannel().send(MessageBuilder.withPayload("bizType.11 message").setHeader("bizType", "bizType.11").build());
        outputChannel.topicRouteOutputChannel().send(MessageBuilder.withPayload("bizType.bb message").setHeader("bizType", "bizType.bb").build());
    }

    /**
     * 发送普通消息（生产者手动ack）
     */
    public void sendConfirmMessage(String msg) {
        //交换机ack成功，路由队列成功
        outputChannel.confirmOutputChannel().send(MessageBuilder.withPayload(msg + "++++").setHeader("bizType", "confirm").build());

        //交换机ack成功，路由队列失败
        outputChannel.confirmOutputChannel().send(MessageBuilder.withPayload(msg + "----").setHeader("bizType", "bizType1").build());
    }

    @ServiceActivator(inputChannel = "confirmAckSuccessChannel")
    public void confirmSuccess(Message<?> message) {
        System.out.println("===============confirm ack success:  " + message);
    }

    @ServiceActivator(inputChannel = "destination.confirm-ack.errors")
    public void confirmFail(Message<?> message) {
        System.out.println("===============confirm ack fail or routing fail:  " + message);
    }
}
